package com.fjec.compensate.service.impl;

import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.core.conditions.update.UpdateWrapper;
import com.fjec.common.Const;
import com.fjec.common.enums.EnumConsumeStatus;
import com.fjec.common.protocol.FjecFruitsProto;
import com.fjec.common.util.Base64Util;
import com.fjec.common.util.DateUtil;
import com.fjec.compensate.entity.BizMqCompensate;
import com.fjec.compensate.entity.BizMqCompensateErr;
import com.fjec.compensate.mapper.BizMqCompensateErrMapper;
import com.fjec.compensate.mapper.BizMqCompensateMapper;
import com.fjec.compensate.service.IBizMqCompensateService;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import lombok.extern.slf4j.Slf4j;
import org.slf4j.MDC;
import org.springframework.amqp.core.Message;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;

/**
 * <p>
 * 019.消息队列补偿表 服务实现类
 * </p>
 *
 * @author admin
 * @since 2020-10-19
 */
@Slf4j
@Service
public class BizMqCompensateServiceImpl extends ServiceImpl<BizMqCompensateMapper, BizMqCompensate> implements IBizMqCompensateService {

    @Resource
    private BizMqCompensateMapper mapper;

    @Resource
    private BizMqCompensateErrMapper errMapper;

    @Override
    public boolean handleDelayMessage(Long dataId, Integer msgType, Message message) {
        try {
            // 查询数据库是否已有此消息
            QueryWrapper<BizMqCompensate> queryWrapper = new QueryWrapper<>();
            queryWrapper.eq("DataId", dataId)
                    .eq("MsgType", msgType);
            BizMqCompensate entity = mapper.selectOne(queryWrapper);
            if (entity != null) {
                // check此消息是否已消费成功
                if (entity.getMsgStatus() != null
                        && entity.getMsgStatus().intValue() == EnumConsumeStatus.COMPLETE.getCode()) {
                    // 已成功返回true
                    return true;
                } else {
                    // 未成功，调用次数+1，并调用RPC重发消息，如果调用次数大于等于3则停止处理，更新为消费失败
                    Long id = entity.getId();
                    int times = entity.getMsgTimes();
                    if (times >= 3) {
                        while (true) {
                            entity = mapper.selectById(id);
                            if (entity.getMsgStatus().intValue() == EnumConsumeStatus.COMPLETE.getCode()) {
                                break;
                            } else {
                                BizMqCompensate updateEntity = new BizMqCompensate();
                                updateEntity.setId(id);
                                updateEntity.setUpdateVer(entity.getUpdateVer());
                                updateEntity.setMsgStatus(EnumConsumeStatus.FAIL.getCode());
                                updateEntity.setUpdateTime(DateUtil.nowLdt());
                                int num = mapper.updateById(updateEntity);
                                if (num >= 1) {
                                    break;
                                }
                            }
                        }
                    } else {
                        // 调用次数+1，并调用RPC重发消息
                        boolean callPpc;
                        while (true) {
                            entity = mapper.selectById(id);
                            if (entity.getMsgStatus().intValue() == EnumConsumeStatus.COMPLETE.getCode()) {
                                callPpc = false;
                                break;
                            } else {
                                BizMqCompensate updateEntity = new BizMqCompensate();
                                updateEntity.setId(id);
                                updateEntity.setUpdateVer(entity.getUpdateVer());
                                updateEntity.setMsgTimes(entity.getMsgTimes() + 1);
                                updateEntity.setUpdateTime(DateUtil.nowLdt());
                                int num = mapper.updateById(updateEntity);
                                if (num >= 1) {
                                    callPpc = true;
                                    break;
                                }
                            }
                        }
                        if (callPpc) {
                            // TODO 调用RPC重发消息
                        }
                    }
                }
            } else {
                // 未成功，落库，并调用RPC重发消息
                this.saveEntity(dataId, msgType, message, EnumConsumeStatus.UNCONSUME.getCode());
                // TODO 调用RPC重发消息
            }
            return true;
        } catch (Exception ex) {
            // 保存处理异常数据
            this.saveErr(dataId, msgType, message, ex.getMessage());
            // 记录日志
            MDC.put(Const.TRACE_ID_KEY, "MQ_COMPLETE_ERR");
            log.error(ex.getMessage(), ex);
            MDC.remove(Const.TRACE_ID_KEY);
            return false;
        }
    }

    @Override
    public boolean handleCompleteMessage(Long dataId, Integer msgType, Message message) {
        try {
            QueryWrapper<BizMqCompensate> queryWrapper = new QueryWrapper<>();
            queryWrapper.eq("DataId", dataId)
                    .eq("MsgType", msgType);
            BizMqCompensate entity = mapper.selectOne(queryWrapper);
            if (entity == null) {
                this.saveEntity(dataId, msgType, message, EnumConsumeStatus.COMPLETE.getCode());
            } else {
                BizMqCompensate updateEntity = new BizMqCompensate();
                updateEntity.setId(entity.getId());
                updateEntity.setUpdateVer(entity.getUpdateVer());
                updateEntity.setMsgStatus(EnumConsumeStatus.COMPLETE.getCode());
                updateEntity.setUpdateTime(DateUtil.nowLdt());
                while (true) {
                    int result = mapper.updateById(updateEntity);
                    if (result >= 1) {
                        break;
                    }
                }
            }
            return true;
        } catch (Exception ex) {
            // 保存处理异常数据
            this.saveErr(dataId, msgType, message, ex.getMessage());
            // 记录日志
            MDC.put(Const.TRACE_ID_KEY, "MQ_COMPLETE_ERR");
            log.error(ex.getMessage(), ex);
            MDC.remove(Const.TRACE_ID_KEY);
            return false;
        }
    }

    /**
     * 保存消息
     *
     * @param dataId
     * @param msgType
     * @param message
     * @param msgStatus
     */
    private void saveEntity(Long dataId, Integer msgType, Message message, Integer msgStatus) {
        BizMqCompensate entity = new BizMqCompensate();
        entity.setDataId(dataId);
        entity.setMsgType(msgType);
        entity.setMsgStatus(msgStatus);
        entity.setContent(message.getBody());
        entity.setCreateTime(DateUtil.nowLdt());
        entity.setMsgTimes(1);
        mapper.insert(entity);
    }

    /**
     * 保存处理异常数据
     *
     * @param dataId
     * @param msgType
     * @param message
     * @param ex
     */
    private void saveErr(Long dataId, Integer msgType, Message message, String ex) {
        BizMqCompensateErr entity = new BizMqCompensateErr();
        entity.setDataId(dataId);
        entity.setMsgType(msgType);
        entity.setMsgStatus(EnumConsumeStatus.COMPLETE.getCode());
        entity.setContent(message.getBody());
        entity.setCreateTime(DateUtil.nowLdt());
        entity.setMsgTimes(1);
        entity.setRemark(ex);
        errMapper.insert(entity);
    }
}
